/* Copyright [2023] <gyl> */
#include "sql/operator/order_physical_operator.h"

RC OrderPhysicalOperator::open(Trx *trx) {
  if (children_.size() > 1) {
    LOG_WARN("order operator must has one child");
    return RC::INTERNAL;
  }
  return children_[0]->open(trx);
}

bool OrderPhysicalOperator::cmp(Tuple *t1, Tuple *t2) {
  for (int i = 0; i < orders_.size(); i++) {
    const char *relation_name = orders_[i].rel.relation_name.c_str();
    const char *attr_name = orders_[i].rel.attribute_name.c_str();
    Value tmp1, tmp2;
    t1->find_cell(TupleCellSpec(relation_name, attr_name), tmp1);
    t2->find_cell(TupleCellSpec(relation_name, attr_name), tmp2);
    if (tmp1.attr_type() == NULLS && tmp2.attr_type() != NULLS) {
      return orders_[i].flag ? false : true;
    } else if (tmp2.attr_type() == NULLS && tmp1.attr_type() != NULLS) {
      return orders_[i].flag ? true : false;
    } else if (tmp1.attr_type() == NULLS && tmp2.attr_type() == NULLS) {
      continue;
    }
    int result = 0;
    tmp1.compare(tmp2, result);
    if (result != 0) {
      if (orders_[i].flag == 0) {
        return result < 0;
      } else {
        return result > 0;
      }
    }
  }
  return false;
}

RC OrderPhysicalOperator::next() {
  if (cnt != 0 && cnt == tuples_.size()) {
    return RC::RECORD_EOF;
  }
  // 下面的代码只会在cnt == 0时被执行一次
  if (cnt == 0) {
    RC rc = RC::SUCCESS;
    cnt++;
    PhysicalOperator *oper = children_.front().get();

    while (RC::SUCCESS == (rc = oper->next())) {
      Tuple *tuple = oper->current_tuple()->copy_record_tuple();
      tuples_.emplace_back(tuple);
    }

    stable_sort(tuples_.begin(), tuples_.end(),
                std::bind(&OrderPhysicalOperator::cmp, this,
                          std::placeholders::_1, std::placeholders::_2));
    if (rc != RC::RECORD_EOF && rc != RC::SUCCESS || tuples_.size() == 0) {
      return rc;
    } else {
      return RC::SUCCESS;
    }
  }
  cnt++;
  return RC::SUCCESS;
}

Tuple *OrderPhysicalOperator::current_tuple() { return tuples_[cnt - 1]; }

RC OrderPhysicalOperator::close() {
  if (!children_.empty()) {
    children_[0]->close();
  }
  cnt = 0;
  return RC::SUCCESS;
}
